package cn.lsh.reducer;

import cn.lsh.writable.Weather;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class WeatherReduce extends Reducer<Weather, IntWritable, Text, IntWritable> {
	private final Text outKey = new Text();

	private final IntWritable wd = new IntWritable();

	@Override
	protected void reduce(Weather key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
		//处理的数据为一组同月的数据，年月顺序，温度倒序；前面为key，后面为value
		// 2020 3 4 38		38
		// 2020 3 6 37		37
		// 2020 3 6 36		36
		boolean flag = true;
		int day = 0;
		for (IntWritable iw : values) {
			//迭代vlaue，key的值会自动对应，ReduceContextImpl内部实现了value和key的同步迭代。
			if (flag) {
				flag = false;
				day = key.getDay();
				//输出第一个值，就是当月温度最高的
				outKey.set(key.toString());
				wd.set(key.getWd());
				context.write(outKey, wd);
			} else if (day != key.getDay()) {
				//输出温度第二高的天
				outKey.set(key.toString());
				wd.set(key.getWd());
				context.write(outKey, wd);
				break;
			}
		}

	}
}
